package cn.doitedu.day02

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object T11_CombineByKeyDemo {

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf
    val conf = new SparkConf().setAppName("MapPartitionsWithIndexDemo")
      .setMaster("local[4]")

    //2.创建SparkContext
    val sc = new SparkContext(conf)

    val lst = List(
      ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
      ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
      ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
      ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
    )
    //通过并行化的方式创建RDD，分区数量为4
    val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)

    //使用combineByKey实现reduceByKey的功能
    //在上游一个分区内，相同的的key第一次出现，value的处理逻辑
    val f1 = (v: Int) => v
    //在上游一个分区内，相同的的key再次出现，对应的value的处理逻辑
    val f2 = (c: Int, v: Int) => c + v
    //在下游，将多个分区，相同key的数据通过网络拉去过来，讲value进行全局聚合的逻辑
    val f3 = (c1: Int, c2: Int) => c1 + c2
    val reduced = wordAndOne.combineByKey(f1, f2, f3)
    reduced.saveAsTextFile("out/out15")
  }

}